package com.tn.rtdata.examples.stockticker.server.util;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Set;

import com.tangosol.net.NamedCache;
import com.tangosol.util.MapEvent;
import com.tangosol.util.MapListener;
import com.tn.rtdata.DataEvent;
import com.tn.rtdata.DataListener;
import com.tn.rtdata.util.ObservableMap;

/**
 * An implementation of <code>ObservableMap</code> that wraps a Coherence <code>NamedCache</code>.
 */
public class CoherenceObservableMap<K, V> implements ObservableMap<K, V>
{
  private List<DataListener<V>> dataListeners;
  private NamedCache namedCache;

  /**
   * Creates a new <code>CoherenceObservableMap</code> initialized with the <code>namedCache</code>.
   */
  public CoherenceObservableMap(NamedCache namedCache)
  {
    this.namedCache = namedCache;
    this.namedCache.addMapListener(new ListenerFiringMapListener());
    this.dataListeners = new ArrayList<DataListener<V>>();
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public int size()
  {
    return namedCache.size();
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public boolean isEmpty()
  {
    return namedCache.isEmpty();
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public boolean containsKey(Object key)
  {
    return namedCache.containsKey(key);
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public boolean containsValue(Object value)
  {
    return namedCache.containsValue(value);
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public V get(Object key)
  {
    //noinspection unchecked
    return (V)namedCache.get(key);
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public V put(K key, V value)
  {
    //noinspection unchecked
    return (V)namedCache.put(key, value);
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public V remove(Object key)
  {
    //noinspection unchecked
    return (V)namedCache.remove(key);
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public void putAll(Map<? extends K, ? extends V> map)
  {
    //noinspection unchecked
    namedCache.putAll(map);
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public void clear()
  {
    namedCache.clear();
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public Set<K> keySet()
  {
    //noinspection unchecked
    return namedCache.keySet();
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public Collection<V> values()
  {
    //noinspection unchecked
    return namedCache.values();
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public Set<Entry<K, V>> entrySet()
  {
    //noinspection unchecked
    return namedCache.entrySet();
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public void addDataListener(DataListener<V> dataListener)
  {
    dataListeners.add(dataListener);
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public void removeDataListener(DataListener<V> dataListener)
  {
    dataListeners.remove(dataListener);
  }

  /**
   * An implementation of <code>MapListener</code> that fires the data listeners.
   */
  private class ListenerFiringMapListener implements MapListener
  {
    /**
     * {@inheritDoc}
     */
    @Override
    public void entryInserted(MapEvent mapEvent)
    {
      //noinspection unchecked
      fireDataListeners(new DataEvent(DataEvent.Type.ADD, mapEvent.getNewValue(), null));
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void entryUpdated(MapEvent mapEvent)
    {
      //noinspection unchecked
      fireDataListeners(new DataEvent(DataEvent.Type.UPDATE, mapEvent.getNewValue(), mapEvent.getOldValue()));
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void entryDeleted(MapEvent mapEvent)
    {
      //noinspection unchecked
      fireDataListeners(new DataEvent(DataEvent.Type.DELETE, null, mapEvent.getOldValue()));
    }

    /**
     * Fires the <code>dataListeners</code>.
     */
    private void fireDataListeners(DataEvent<V> dataEvent)
    {
      for (DataListener<V> dataListener : new ArrayList<DataListener<V>>(dataListeners))
      {
        dataListener.dataEvent(dataEvent);
      }
    }
  }
}
